package org.example.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  案例实操
 *
 *  统计出每一个**省份**每个广告被点击数量排行的Top3
 */
object Spark21_RDD_Operator_Req {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)

    // 获取原始数据
    val dataRdd: RDD[String] = context.textFile("file.log")

    // 原始数据进行转换
    // 时间戳，省份，城市，用户 广告
    // => （ （省份，广告），1）
    // => （ （省份，广告），1）=》（ （省份，广告），sum）
    // =》 （ 省份，（广告，sum））
    // => （ 省份，[（广告A，sum）,（广告B，sum）]）
    // =》 将分组后的数据进行排序，并取前三名

    val reby: RDD[((String, String), Int)] = dataRdd.map(row => {
      val rows: Array[String] = row.split(" ")
      ((rows(1), rows(4)), 1)
    }).reduceByKey(_ + _) // 将值进行聚合,
    val changRdd: RDD[(String,( String, Int))] =
      reby.map{
        case ((shenfe,guanggao),sum) =>{
          (shenfe,(guanggao,sum))
        }
      }
    // 转换 （ 省份，（广告，sum））
    // 进行分组
    val groupRdd: RDD[(String, Iterable[(String, Int)])] =
    changRdd.groupByKey()

    val resutlRdd: RDD[(String, List[(String, Int)])] = groupRdd.mapValues(
      iter => {
        // 迭代器不支持排序，所以转换成list
        // scala的集合中，默认是升序，所以这里设置成降序，并且只取三个
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )
    resutlRdd.collect().foreach(println)


    // 采集数据打印

    context.stop()
  }
}
